package producer

import (
	"context"
	"encoding/json"

	"github.com/IBM/sarama"
)

// 生产者要发送的消息
type Event interface {
	// Topic() string
	Key() string
}

type Producer struct {
	producer sarama.SyncProducer
	topic    string
}

func NewSyncProducer(client sarama.Client, topic string) (*Producer, error) {
	p, err := sarama.NewSyncProducerFromClient(client)
	if err != nil {
		return nil, err
	}
	return &Producer{
		producer: p,
		topic:    topic,
	}, nil
}

func (s *Producer) ProduceEvent(ctx context.Context, evt Event) error {
	data, err := json.Marshal(evt)
	if err != nil {
		return err
	}
	msg := sarama.ProducerMessage{
		Topic: s.topic,
		Value: sarama.ByteEncoder(data),
	}
	if len(evt.Key()) > 0 {
		msg.Key = sarama.StringEncoder(evt.Key())
	}
	_, _, err = s.producer.SendMessage(&msg)
	return err
}
